package com.imooc.bilibili.service.websocket;

import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.config.ConfirmConfig;
import com.imooc.bilibili.domain.Danmu;
import com.imooc.bilibili.service.DanmuService;
import com.imooc.bilibili.util.TokenUtil;
import com.mysql.cj.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@ServerEndpoint("/imserver/{token}")
public class WebSocketService {
    @Autowired
    private RabbitTemplate rabbitTemplate;


    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    // 线程安全整数型包装类
    public static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    // 线程安全hashmap
    public static final ConcurrentHashMap<String,WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    private Session session;

    private Long userId;

    private String sessionId;

    // @Autowired
    // private StringRedisTemplate redisTemplate;

    public static ApplicationContext APPLICATION_CONTEXT;

    public static void setApplicationContext(ApplicationContext applicationContext){
        WebSocketService.APPLICATION_CONTEXT = applicationContext;
    }


    @OnOpen
    public void openConnection(Session session, @PathParam("token") String token){
        try {
            this.userId = TokenUtil.verifyToken(token);
        } catch (Exception e) {}
        this.sessionId = session.getId();
        this.session = session;
        if (WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId,this);
        }else {
            WEBSOCKET_MAP.put(sessionId,this);
            ONLINE_COUNT.getAndIncrement();
        }

        logger.info("用户连接成功：{}，当前在线人数为：{}",sessionId,ONLINE_COUNT.get());
        try {
            this.sendMessage("0");
        } catch (Exception e) {
            logger.error("连接异常！");
        }
    }

    @OnClose
    public void closeConnection(){
            if(WEBSOCKET_MAP.containsKey(sessionId)){
                WEBSOCKET_MAP.remove(sessionId);
                ONLINE_COUNT.getAndDecrement();
            }

        logger.info("用户退出:{},当前在线人数为：{}",sessionId,ONLINE_COUNT.get());
    }

    /*@OnMessage
    public void onMessage(String message) {
        logger.info("用户信息：{}，报文：{}",sessionId,message);
        if (!StringUtils.isNullOrEmpty(message)){
            try {
                for (Map.Entry<String,WebSocketService> entry : WEBSOCKET_MAP.entrySet()){
                    WebSocketService webSocketService = entry.getValue();
                    DefaultMQProducer danmuProducer = (DefaultMQProducer) APPLICATION_CONTEXT.getBean("danmuProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message",message);
                    jsonObject.put("sessionId",webSocketService.getSessionId());
                    Message msg = new Message(UserMomentsConstant.TOPIC_DANMUS, jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    RocketMQUtil.asyncSendMsg(danmuProducer,msg);
                }
                if (userId != null){
                    // 将弹幕保存到数据库
                    Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                    danmu.setUserId(userId);
                    danmu.setCreateTime(new Date());
                    DanmuService danmuService = (DanmuService)APPLICATION_CONTEXT.getBean("danmuService");
                    danmuService.asyncAddDanmu(danmu);
                    // 将弹幕保存到redis
                    danmuService.addDanmusToRedis(danmu);
                }
            } catch (Exception e) {
                logger.error("弹幕接受异常！");
                e.printStackTrace();
            }
        }
    }*/
    @OnMessage
    public void onMessage(String message) {
        logger.info("用户信息：{}，报文：{}",sessionId,message);
        if (!StringUtils.isNullOrEmpty(message)){
            try {
                for (Map.Entry<String,WebSocketService> entry : WEBSOCKET_MAP.entrySet()){
                    WebSocketService webSocketService = entry.getValue();
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message",message);
                    jsonObject.put("sessionId",webSocketService.getSessionId());
                    CorrelationData correlationData = new CorrelationData("1");
                    Message msg = new Message(jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    RabbitTemplate rabbitTemplate = (RabbitTemplate) APPLICATION_CONTEXT.getBean("rabbitTemplate");
                    rabbitTemplate.convertAndSend(ConfirmConfig.DANMU_EXCHANGE_NAME,
                            ConfirmConfig.DANMU_ROUTING_KEY,msg,correlationData);
                }
                if (this.userId != null){
                    // 将弹幕保存到数据库
                    Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                    danmu.setUserId(userId);
                    danmu.setCreateTime(new Date());
                    DanmuService danmuService = (DanmuService)APPLICATION_CONTEXT.getBean("danmuService");
                    danmuService.asyncAddDanmu(danmu);
                    // 将弹幕保存到redis
                    danmuService.addDanmusToRedis(danmu);
                }
            } catch (Exception e) {
                logger.error("弹幕接受异常！");
                e.printStackTrace();
            }
        }
    }

    @Scheduled(fixedRate = 5000)
    public void noticeOnlineCount() throws IOException {
        for (Map.Entry<String,WebSocketService> entry : WEBSOCKET_MAP.entrySet()){
            WebSocketService webSocketService = entry.getValue();
            if (webSocketService.getSession().isOpen()){
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onlineCount",ONLINE_COUNT.get());
                jsonObject.put("msg","在线人数为："+ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());
            }
        }
    }
    @OnError
    public void onError(Throwable error){

    }

    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }
}
